package cn.genmer.test.security.io.testreactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多路复用器 线程组
 */
public class SelectorThreadGroup {
    SelectorThread[] threadArray;
    ServerSocketChannel server = null;
    private AtomicInteger xid = new AtomicInteger(0);

    SelectorThreadGroup workerGroup = this;

    public void setWorkerGroup(SelectorThreadGroup workerGroup){
        this.workerGroup = workerGroup;
    }


    SelectorThreadGroup(int num){
        // num为1, 一个线程负责接收处理
        // num为3, 混杂模式 一个线程负责accept, 然后每个线程负责接收分配的client, 进行R/W（包括负责accept的）
        threadArray = new SelectorThread[num];
        for (int i = 0; i < num; i++) {
            threadArray[i] = new SelectorThread(this);
            // 立即启动线程
            new Thread(threadArray[i]).start();
        }

    }

    void bind(int port){
        try {
            // 创建
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));
            // 注册
            nextSelectorV3(server);
//            nextSelectorV2(server);
            
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * serverSocket socket复用这个方法
     * @param server
     */
    public void nextSelectorV3(Channel server) {
        SelectorThread st;
        try {
            if (server instanceof ServerSocketChannel){
                st = next();
                // 通过队列传递消息
                st.lbq.put(server);
                // boss中选择一个线程后，更新指向的work组（默认是本身）
                st.setWorker(workerGroup);
                // 打断server()方法阻塞， 让eventLoop进入第三步 run all task
                st.selector.wakeup();
            } else  {
                st = nextV3(); // 在main线程中，取到堆中的selectorThread对
                // 通过队列传递消息
                st.lbq.add(server);
                // 打断server()方法阻塞， 让eventLoop进入第三步 run all task
                st.selector.wakeup();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public void nextSelectorV2(Channel server) {
        SelectorThread st;
        if (server instanceof ServerSocketChannel){
            st = threadArray[0];
            st.selector.wakeup();
        } else  {
            st = nextV2(); // 在main线程中，取到堆中的selectorThread对
        }

        // 通过队列传递消息
        st.lbq.add(server);
        // 打断server()方法阻塞， 让eventLoop进入第三步 run all task
        st.selector.wakeup();
    }
    public void nextSelector(Channel server) {
        SelectorThread st = next(); // 在main线程中，取到堆中的selectorThread对
        // 通过队列传递消息
        st.lbq.add(server);
        // 打断server()方法阻塞， 让eventLoop进入第三步 run all task
        st.selector.wakeup();
    }
    /**
     * serverSocket socket复用这个方法
     * 这个直接调用其他线程的方法，容易被阻塞住（虽然我测试都是正常可以接收的）
     * @param server
     */
    private void nextSelectorOld(Channel server) {
        SelectorThread st = next();

        ServerSocketChannel channel = (ServerSocketChannel) server;
        // 呼应上 SelectorThreadGroup在被实例化的时候, sts里线程在创建后就立刻启动了，所以此时， st（选中的这个selector）是阻塞的
        try {
            // 使用wakeup让selector的select()方法立刻返回, 不阻塞.
            st.selector.wakeup();
            channel.register(st.selector, SelectionKey.OP_ACCEPT); // 不使用wakeup, 会被阻塞！！！

        } catch (ClosedChannelException e) {
            e.printStackTrace();
        }
    }

    /**
     * 轮询获取组内线程
     * @return
     */
    private SelectorThread next(){
        //TODO 可以在这个基础上记录下调用次数, 来进行一个轮询平均点的优化
        int index = xid.incrementAndGet() % threadArray.length;
        return threadArray[index];
    }
    private SelectorThread nextV2(){
        // accept专门accept，不分配客户端 - 选用
        int index = xid.incrementAndGet() % (threadArray.length-1);
        return threadArray[index+1];
    }
    private SelectorThread nextV3(){
        //TODO 可以在这个基础上记录下调用次数, 来进行一个轮询平均点的优化
        int index = xid.incrementAndGet() % workerGroup.threadArray.length;
        return workerGroup.threadArray[index];
    }
}
